package com.foruo.algo.bucket;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 漏桶算法
 *
 * 1、漏桶以恒定的速率流出（调用后续方法）
 * 2、漏桶有最大容量上限，会溢出
 * 3、先进入桶的先执行
 *
 * @author GaoYuan
 * @date 2018/10/25 上午10:55
 */
public class BucketLeakyAlgo {

    /** 请求列表 - 拥堵数据 */
    static volatile Queue<String> requestQueue = new ConcurrentLinkedDeque<>();

    /** 每秒允许执行10个数据 */
    static long passNumPerSecond = 10;

    /** 桶允许的最大值 */
    static long bucketNumMax = 100;

    /** 当前桶的数据量 */
    static volatile int bucketNumCurrentInteger = 0;

    /** 记录每次流出去的时间，只保留一秒内的数据 */
    static volatile List<Long> perOutRecord = new CopyOnWriteArrayList<>();

    /** 统计错误量 */
    static volatile AtomicInteger errors = new AtomicInteger(0);


    /**
     * 获取
     * false 表示 桶满了
     * true 表示成功入桶
     * @author GaoYuan
     * @date 2018/10/25 上午11:06
     */
    public static synchronized boolean tryAquire(){
        // 判断桶还能不能进去
        if (bucketNumCurrentInteger > bucketNumMax) {
            // 桶满了
            errors.incrementAndGet();
            return false;
        }
        bucketNumCurrentInteger++;
        // 清除一秒外的数据
        clearPerOutRecord();
        // 添加到requestQueue中排队
        requestQueue.add(Thread.currentThread().getName());
        // 目前1秒内已经有10个流量出去了，需要循环等待
        while (perOutRecord.size() >= passNumPerSecond) {
            // 清除一秒外的数据
            clearPerOutRecord();
        }
        // 轮到自己，才能调用
        while (!Thread.currentThread().getName().equals(requestQueue.peek())){}
        requestQueue.poll();
        // 说明单位时间还能执行
        perOutRecord.add(System.currentTimeMillis());
        // 减少当前水量
        bucketNumCurrentInteger--;
        return true;
    }

    /** 清除一秒外的数据 */
    public static void clearPerOutRecord(){
        long nowTm = System.currentTimeMillis();
        Iterator iterator = perOutRecord.iterator();
        List<Long> newList = new CopyOnWriteArrayList<>();
        while (iterator.hasNext()) {
            long time = (Long) iterator.next();
            if ((nowTm - time) < 1000) {
                newList.add(time);
            }
        }
        perOutRecord = newList;
    }

    public static class MyThread extends Thread{

        private int i;

        public MyThread(int i){
            this.i = i;
        }

        @Override
        public void run() {
            try{
                long nowTm = System.currentTimeMillis();
                if(tryAquire()){
                    System.out.println("i = "+ i +"，" + Thread.currentThread().getName() + "执行成功了，" + nowTm);
                }else{
                    System.err.println("i = "+ i +"，" + Thread.currentThread().getName() + "执行失败了，" + nowTm);
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }

        public int getI() {
            return i;
        }

        public void setI(int i) {
            this.i = i;
        }

    }

    /**
     * 第0轮打印 20 条
     *  其中： 前10条在第一秒打印，后10条是一秒后打印
     * @author GaoYuan
     * @date 2018/10/25 下午5:42
     */
    public static void main(String[] args) throws Exception{
        // 每秒 20 请求
        int index = 0;
        int j=0;
        while (j<2) {
            TimeUnit.SECONDS.sleep(1);
            System.out.println("第" + index ++ + "轮");
            for (int i = 0; i < 15; i++) {
                new MyThread(i).start();
            }
            j++;
            // 第10次一下子请求300次
            if(j==10){
                for (int i = 0; i < 300; i++) {
                    new MyThread(i).start();
                }
            }
        }

        TimeUnit.SECONDS.sleep(5);
        System.out.println("失败数量 = " + errors);
    }

}
